package com.baijiaoxi.common.model.asyncjob.eventbus;

import com.baijiaoxi.common.model.asyncjob.AsyncJobProducer;
import com.baijiaoxi.common.model.base.CoreException;
import com.baijiaoxi.common.model.concurrent.BaiExecutors;
import com.baijiaoxi.common.model.mq.MQBody;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class EventBusManager {
    private Map<String, EventBus> topicEventBusMap = new HashMap<>();
    @Getter
    private EventStorage eventStorage;


    //支持未处理的事件重启时重新执行
    public EventBusManager(EventStorage eventStorage) {
        this.eventStorage = eventStorage;
    }

    public EventBusManager() {

    }

    public EventBus register(EventBusSubscriber eventSubscriber, AsyncJobProducer asyncJobProducer) {
        String topic = eventSubscriber.topic();
        int poolSize = eventSubscriber.poolSize();
        log.info("注册topic:{},处理线程数:{}", topic, poolSize);
        EventBus eventBus = new AsyncEventBus(topic, BaiExecutors.create(topic, poolSize));
        topicEventBusMap.put(topic, eventBus);
        eventBus.register(eventSubscriber);
        republishStorageEvent(topic, asyncJobProducer);
        return eventBus;
    }

    private void republishStorageEvent(String topic, AsyncJobProducer asyncJobProducer) {
        if (eventStorage == null)
            return;
        Map<String, MQBody> storageMqBodyMap = eventStorage.getAll(topic);
        if (storageMqBodyMap == null || storageMqBodyMap.isEmpty())
            return;
        storageMqBodyMap.entrySet().forEach(entry -> {
            asyncJobProducer.republish(topic, entry.getValue());
        });
    }

    EventBus getEventBus(String topic) {
        if (topicEventBusMap.containsKey(topic))
            return topicEventBusMap.get(topic);
        throw new CoreException(topic + "事件未注册");
    }
}
